package mr

import (
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"path"
	"path/filepath"
	"runtime"
	"sort"
	"strings"
	"time"
)
import "log"
import "net/rpc"
import "hash/fnv"

//
// Map functions return a slice of KeyValue.
//
type KeyValue struct {
	Key   string
	Value string
}

//
// use ihash(key) % NReduce to choose the reduce
// Task number for each KeyValue emitted by Map.
//
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

var infoBody InitInfoBody

var pwd string

var outDir, mainDir string = func() (string, string) {
	pwd, _ = os.Getwd()
	oDir := pwd
	var mDir string
	if runtime.GOOS == "linux" {
		mDir = oDir[:strings.LastIndex(pwd, "main")+3]
	} else if runtime.GOOS == "darwin" {
		if strings.Contains(pwd, "main") {
			mDir = pwd
			oDir = path.Join(filepath.Dir(pwd), "out")
		} else {
			mDir = path.Join(oDir, "main")
		}
	}
	return oDir, mDir
}()

//
// main/mrworker.go calls this function.
//
func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {
	// Your worker implementation here.
	// 连接coordinate
	conn, err := rpc.DialHTTP("unix", coordinatorSock())
	if err != nil {
		log.Fatal(err)
		return
	}

	//pwd, _ = os.Getwd()

	var req []interface{}
	res := ResponseBody{}

	if err = conn.Call("Coordinator.WorkerInit", req, &res); err != nil || res.Type_id != RES_INFO {
		log.Fatal(err)
		return
	} else {
		err := json.Unmarshal([]byte(res.Content), &infoBody)
		if err != nil {
			log.Panic("Worker", infoBody.WorkerId, ": ", "failed to parse init json data, error: ", err)
		}
	}

	// 循环等待coordinate的回应
	// 		若成功收到了任务，回应，开始工作
	// 		做完后，发送消息（包括中间/结果文件的地址发过去），继续等待回应：
	// 		若收到结束/中止的消息，结束

	req = append(req, infoBody.WorkerId)

	var workChan chan error
	sigChan := make(chan byte)
	go func() {
		for {
			<-sigChan
			res = ResponseBody{}
			e := conn.Call("Coordinator.NextOperation", req, &res)
			workChan <- e
		}
	}()

	var workErr error

	handleWorkErr := func() {
		if workErr != nil {
			// 得到了任务过程中产生的错误，即任务失败，返回失败消息
			//log.Println("Worker", infoBody.WorkerId, ": ", "Task failed")
			var info []interface{}
			info = append(append(info, infoBody.WorkerId), "failed")
			err = conn.Call("Coordinator.TaskWorkResDispatch", info, nil)
		}
	}

	for {
		// 获取下一个操作，超时便退出
		res = ResponseBody{}
		workChan = make(chan error)
		sigChan <- 1

		timer := time.NewTimer(WAIT_TIME)

		select {
		case err, ok := <-workChan:
			if err != nil || !ok {
				close(workChan)
				if err == io.EOF || err.Error() == "unexpected EOF" {
					log.Println("Worker", infoBody.WorkerId, ": ", "is going to close because channel of next op is closed")
					idInfo := make([]interface{}, 1)
					idInfo[0] = infoBody.WorkerId
					_ = conn.Call("Coordinator.WorkerWannaClose", idInfo, nil)
					return
				}
				log.Panic("Worker", infoBody.WorkerId, ": ", "failed to get next operation, error is ", err)
			}
		case <-timer.C:
			idInfo := make([]interface{}, 1)
			idInfo[0] = infoBody.WorkerId
			_ = conn.Call("Coordinator.WorkerWannaClose", idInfo, nil)
			close(workChan)
			log.Panic("Worker", infoBody.WorkerId, ": ", "expired in getting next operation")
		}
		close(workChan)
		timer.Stop()

		switch res.Type_id {
		case RES_INFO:
			log.Println("Worker", infoBody.WorkerId, ":", "got info:", res)
			switch res.Desc {
			case "waiting":
				time.Sleep(1 * time.Second)
			}
		case RES_TASK:
			var task Task
			err := json.Unmarshal([]byte(res.Content), &task)
			if err != nil {
				log.Panic("Worker", infoBody.WorkerId, ": ", "failed to parse map Task json data, error: ", err)
			}
			switch task.TaskType {
			// map任务
			case "map":
				// 读出文件的内容
				var content []byte
				filePath, _ := filepath.Rel(pwd, task.Content[0])
				content, workErr = ioutil.ReadFile(filePath)
				if workErr != nil {
					handleWorkErr()
					log.Println("Worker", infoBody.WorkerId, ": ", "Read Map File Error: ", workErr)
					continue
				}
				// 返回kv对，key是单词，value是统计内容，wc中是“1”，代表1个
				kv := mapf(filePath, string(content))
				fp, we := doAfterMap(&task, &kv, infoBody.NReduce)
				if we != nil {
					workErr = we
					handleWorkErr()
					log.Println("Worker", infoBody.WorkerId, ": ", "doAfterMap failed, error is ", we)
					continue
				}
				wkr := WorkerRes{int32(infoBody.WorkerId), "done", fp}
				log.Println("Worker", infoBody.WorkerId, ": ", "map task", task.Num, "result:", wkr)
				doneRes := ResponseBody{}
				err = conn.Call("Coordinator.TaskWorkResDispatch", wkr, &doneRes)
				if err != nil {
					log.Panic("Worker", infoBody.WorkerId, ": ", "Reduce rpc return res failed")
				}
				log.Println("Worker", infoBody.WorkerId, ": ", "map task", task.Num, " done")
			// reduce任务
			case "reduce":
				if rkvs, workerErr := doPreReduce(&task.Content); workerErr != nil {
					log.Println(workerErr)
					continue
				} else {
					outName := fmt.Sprintf("mr-out-%d", task.Num)
					outPath := path.Join(outDir, outName)
					outFile, we := os.Create(outPath)
					if we != nil {
						workErr = we
						handleWorkErr()
						log.Println("Worker", infoBody.WorkerId, ": ", "Reduce file created failed")
						continue
					}
					defer outFile.Close()
					for _, rkv := range rkvs {
						resValue := reducef(rkv.key, rkv.values)
						//enc := json.NewEncoder(outFile)
						if _, we = fmt.Fprintf(outFile, "%v %v\n", rkv.key, resValue); we != nil {
							workErr = we
							handleWorkErr()
							log.Println("Worker", infoBody.WorkerId, ": ", "Reduce kv written failed")
							continue
						}
					}
					// rpc发送结果
					wres := WorkerRes{
						int32(infoBody.WorkerId),
						"done",
						[]string{outName},
					}
					err = conn.Call("Coordinator.TaskWorkResDispatch", wres, nil)
					if err != nil {
						log.Panic("Worker", infoBody.WorkerId, ": ", "Reduce rpc return res failed")
					}
					log.Println("Worker", infoBody.WorkerId, ": ", "reduce task", task.Num, " done")
				}
			}
		case RES_CMD:
			switch res.Desc {
			case "close":
				log.Println("Worker", infoBody.WorkerId, ": ", "forced to close")
				return
			}
		}
	}
}

func doPreReduce(filePathArr *[]string) (kvSlice ReduceKVSlice, err error) {
	kvSlice = make(ReduceKVSlice, 0)
	kvIndex := make(map[string]int, 0)
	for i := 0; i < len(*filePathArr); i++ {
		file, e := os.Open((*filePathArr)[i])
		if e != nil {
			err = e
		}
		defer file.Close()
		dec := json.NewDecoder(file)
		var kv KeyValue
		for {
			err = dec.Decode(&kv)
			if err != nil {
				break
			}
			if _, ok := kvIndex[kv.Key]; !ok {
				kvSlice = append(kvSlice, ReduceKVs{kv.Key, []string{}})
				kvIndex[kv.Key] = len(kvSlice) - 1
			}
			kvSlice[kvIndex[kv.Key]].values = append(kvSlice[kvIndex[kv.Key]].values, kv.Value)
		}

		if err != io.EOF {
			return
		}
	}

	sort.Sort(kvSlice)

	err = nil
	return
}

func doAfterMap(task *Task, kvs *[]KeyValue, nReduce int) (fp []string, err error) {
	fp = make([]string, nReduce)
	//fs := make([]*os.File, NReduce)
	encs := make([]*json.Encoder, nReduce)
	for i := 0; i < len(*kvs); i++ {
		kv := (*kvs)[i]
		rd_n := ihash(kv.Key) % nReduce
		if encs[rd_n] == nil {
			//file_name := fmt.Sprintf("mr-temp-%d-%d", strconv.Itoa(Task.Num), rd_n)
			f, e := ioutil.TempFile(outDir, "mr-tmp")
			if e != nil {
				err = e
				log.Panic("Worker", infoBody.WorkerId, ": ", "file ", rd_n, " unable to be created\n")
			}
			defer f.Close()
			encs[rd_n] = json.NewEncoder(f)
			fp[rd_n] = f.Name()
		}
		err = (*encs[rd_n]).Encode(&kv)
		if err != nil {
			log.Panic("Worker", infoBody.WorkerId, ": ", ": write json kv faild\n")
		}
	}
	for i := 0; i < nReduce; i++ {
		if encs[i] == nil {
			continue
		}
		fn := path.Join(outDir, fmt.Sprintf("mr-%d-%d", task.Num, i))
		err = os.Rename(fp[i], fn)
		if err != nil {
			log.Println("Worker", infoBody.WorkerId, ": ", "rename failed at rd_n ", i, "\n")
		} else {
			fp[i] = fn
		}
		//else if err = fs[i].Close(); err != nil {
		//	log.Panic("Worker", infoBody.WorkerId, ": ", "close failed at rd_n ", i, "\n")
		//}
	}

	return
}

//
// send an RPC request to the coordinator, wait for the response.
// usually returns true.
// returns false if something goes wrong.
//
func call(rpcname string, args interface{}, reply interface{}) bool {
	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	sockname := coordinatorSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		log.Fatal("dialing:", err)
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	fmt.Println(err)
	return false
}
